| 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319 |
2x
2x
2x
2x
2x
2x
943x
943x
943x
943x
2x
2x
35x
35x
2x
2x
2x
2x
2x
2x
2x
2x
2052x
2052x
4306x
4488x
2x
2x
721x
721x
721x
721x
721x
721x
721x
2x
2350x
2350x
691x
1659x
1632x
27x
27x
663x
663x
2292x
721x
721x
721x
705x
67x
721x
721x
2x
1675x
1675x
705x
705x
1675x
2x
1930x
2x
691x
691x
623x
610x
610x
610x
691x
85x
84x
84x
84x
691x
673x
1632x
1632x
981x
981x
43x
42x
43x
418x
418x
408x
408x
408x
418x
418x
73x
73x
73x
431x
419x
419x
431x
16x
16x
16x
16x
2x
491x
491x
460x
31x
2x
27x
25x
2x
895x
455x
| /**
* Copyright 2017 Google Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { SnapshotVersion } from '../core/snapshot_version';
import { ProtoByteString, TargetId } from '../core/types';
import { QueryData } from '../local/query_data';
import { maybeDocumentMap } from '../model/collections';
import { Document, NoDocument } from '../model/document';
import { DocumentKey } from '../model/document_key';
import { emptyByteString } from '../platform/platform';
import { assert, fail } from '../util/assert';
import { FirestoreError } from '../util/error';
import * as objUtils from '../util/obj';
import { ExistenceFilter } from './existence_filter';
import {
CurrentStatusUpdate,
RemoteEvent,
ResetMapping,
TargetChange,
UpdateMapping
} from './remote_event';
/**
* Internal representation of the watcher API protocol buffers.
*/
export type WatchChange =
| DocumentWatchChange
| WatchTargetChange
| ExistenceFilterChange;
/**
* Represents a changed document and a list of target ids to which this change
* applies.
*
* If document has been deleted NoDocument will be provided.
*/
export class DocumentWatchChange {
constructor(
/** The new document applies to all of these targets. */
public updatedTargetIds: TargetId[],
/** The new document is removed from all of these targets. */
public removedTargetIds: TargetId[],
/** The key of the document for this change. */
public key: DocumentKey,
/**
* The new document or NoDocument if it was deleted. Is null if the
* document went out of view without the server sending a new document.
*/
public newDoc: Document | NoDocument | null
) {}
}
export class ExistenceFilterChange {
constructor(
public targetId: TargetId,
public existenceFilter: ExistenceFilter
) {}
}
export enum WatchTargetChangeState {
NoChange,
Added,
Removed,
Current,
Reset
}
export class WatchTargetChange {
constructor(
/** What kind of change occurred to the watch target. */
public state: WatchTargetChangeState,
/** The target IDs that were added/removed/set. */
public targetIds: TargetId[],
/**
* An opaque, server-assigned token that allows watching a query to be
* resumed after disconnecting without retransmitting all the data that
* matches the query. The resume token essentially identifies a point in
* time from which the server should resume sending results.
*/
public resumeToken: ProtoByteString = emptyByteString(),
/** An RPC error indicating why the watch failed. */
public cause: FirestoreError | null = null
) {}
}
/**
* A helper class to accumulate watch changes into a RemoteEvent and other
* target information.
*/
export class WatchChangeAggregator {
constructor(
private snapshotVersion: SnapshotVersion,
private readonly listenTargets: { [targetId: number]: QueryData },
pendingTargetResponses: { [targetId: number]: number }
) {
this.pendingTargetResponses = objUtils.shallowCopy(pendingTargetResponses);
}
/** The existence filter - if any - for the given target IDs. */
readonly existenceFilters: { [targetId: number]: ExistenceFilter } = {};
/** The number of pending responses that we are waiting on from watch. */
readonly pendingTargetResponses: { [targetId: number]: number };
/** Keeps track of the current target mappings */
private targetChanges: { [targetId: number]: TargetChange } = {};
/** Keeps track of document to update */
private documentUpdates = maybeDocumentMap();
/** Whether this aggregator was frozen and can no longer be modified */
private frozen = false;
/** Aggregates a watch change into the current state */
add(watchChange: WatchChange): void {
assert(!this.frozen, 'Trying to modify frozen WatchChangeAggregator.');
if (watchChange instanceof DocumentWatchChange) {
this.addDocumentChange(watchChange);
} else if (watchChange instanceof WatchTargetChange) {
this.addTargetChange(watchChange);
} else Eif (watchChange instanceof ExistenceFilterChange) {
this.addExistenceFilterChange(watchChange);
} else {
fail('Unknown watch change: ' + watchChange);
}
}
/** Aggregates all provided watch changes to the current state in order */
addChanges(watchChanges: WatchChange[]): void {
assert(!this.frozen, 'Trying to modify frozen WatchChangeAggregator.');
watchChanges.forEach(change => this.add(change));
}
/**
* Converts the current state into a remote event with the snapshot version
* provided via the constructor.
*/
createRemoteEvent(): RemoteEvent {
const targetChanges = this.targetChanges;
// Remove all the non-active targets from the remote event.
objUtils.forEachNumber(this.targetChanges, targetId => {
if (!this.isActiveTarget(targetId)) {
delete targetChanges[targetId];
}
});
// Mark this aggregator as frozen so no further modifications are made
this.frozen = true;
return new RemoteEvent(
this.snapshotVersion,
targetChanges,
this.documentUpdates
);
}
private ensureTargetChange(targetId: TargetId): TargetChange {
let change = this.targetChanges[targetId];
if (!change) {
// Create an UpdateMapping by default, since resets are always explicit.
change = {
currentStatusUpdate: CurrentStatusUpdate.None,
snapshotVersion: this.snapshotVersion,
mapping: new UpdateMapping(),
resumeToken: emptyByteString()
};
this.targetChanges[targetId] = change;
}
return change;
}
/**
* We need to wait for watch to ack targets before we process those events,
* so to know if a target is active, there must be no pending acks we're
* waiting for and it must be in the current list of targets that the client
* cares about.
*
* This method is visible for testing.
*/
protected isActiveTarget(targetId: TargetId): boolean {
return (
!objUtils.contains(this.pendingTargetResponses, targetId) &&
objUtils.contains(this.listenTargets, targetId)
);
}
private addDocumentChange(docChange: DocumentWatchChange) {
let relevant = false;
for (const targetId of docChange.updatedTargetIds) {
if (this.isActiveTarget(targetId)) {
const change = this.ensureTargetChange(targetId);
change.mapping.add(docChange.key);
relevant = true;
}
}
for (const targetId of docChange.removedTargetIds) {
if (this.isActiveTarget(targetId)) {
const change = this.ensureTargetChange(targetId);
change.mapping.delete(docChange.key);
relevant = true;
}
}
// Only update the document if there is a new document to replace to an
// active target that is being listened to, this might be just a target
// update instead.
if (docChange.newDoc && relevant) {
this.documentUpdates = this.documentUpdates.insert(
docChange.key,
docChange.newDoc
);
}
}
private addTargetChange(targetChange: WatchTargetChange) {
targetChange.targetIds.forEach(targetId => {
const change = this.ensureTargetChange(targetId);
switch (targetChange.state) {
case WatchTargetChangeState.NoChange:
if (this.isActiveTarget(targetId)) {
// Creating the change above satisfies the semantics of no-change.
applyResumeToken(change, targetChange.resumeToken);
}
break;
case WatchTargetChangeState.Added:
// We need to decrement the number of pending acks needed from watch
// for this targetId.
this.recordTargetResponse(targetId);
if (!objUtils.contains(this.pendingTargetResponses, targetId)) {
// We have a freshly added target, so we need to reset any state
// that we had previously This can happen e.g. when remove and add
// back a target for existence filter mismatches.
change.mapping = new UpdateMapping();
change.currentStatusUpdate = CurrentStatusUpdate.None;
delete this.existenceFilters[targetId];
}
applyResumeToken(change, targetChange.resumeToken);
break;
case WatchTargetChangeState.Removed:
// We need to keep track of removed targets to we can
// post-filter and remove any target changes.
// We need to decrement the number of pending acks needed from watch
// for this targetId.
this.recordTargetResponse(targetId);
assert(
!targetChange.cause,
'WatchChangeAggregator does not handle errored targets'
);
break;
case WatchTargetChangeState.Current:
if (this.isActiveTarget(targetId)) {
change.currentStatusUpdate = CurrentStatusUpdate.MarkCurrent;
applyResumeToken(change, targetChange.resumeToken);
}
break;
case WatchTargetChangeState.Reset:
Eif (this.isActiveTarget(targetId)) {
// Overwrite any existing target mapping with a reset
// mapping. Every subsequent update will modify the reset
// mapping, not an update mapping.
change.mapping = new ResetMapping();
applyResumeToken(change, targetChange.resumeToken);
}
break;
default:
fail('Unknown target watch change state: ' + targetChange.state);
}
});
}
/**
* Record that we get a watch target add/remove by decrementing the number of
* pending target responses that we have.
*/
private recordTargetResponse(targetId: TargetId): void {
const newCount = (this.pendingTargetResponses[targetId] || 0) - 1;
if (newCount === 0) {
delete this.pendingTargetResponses[targetId];
} else {
this.pendingTargetResponses[targetId] = newCount;
}
}
private addExistenceFilterChange(change: ExistenceFilterChange): void {
if (this.isActiveTarget(change.targetId)) {
this.existenceFilters[change.targetId] = change.existenceFilter;
}
}
}
/**
* Applies the resume token to the TargetChange, but only when it has a new
* value. null and empty resumeTokens are discarded.
*/
function applyResumeToken(
change: TargetChange,
resumeToken: ProtoByteString
): void {
if (resumeToken.length > 0) {
change.resumeToken = resumeToken;
}
}
|